package com.zshield.producer;

import com.zshield.config.KafkaConfig;
import com.zshield.run.KafkaPreCompute;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.log4j.Logger;

public class CustomKafkaProducer {
    private final KafkaProducer<String, String> producer;
    private static CustomKafkaProducer pro = null;
    private static final Logger logger = Logger.getLogger(CustomKafkaProducer.class);

    private CustomKafkaProducer() {
        this.producer = new KafkaProducer<>(KafkaConfig.getProducerConfig());
    }

    public static CustomKafkaProducer getInstance() {
        if (pro == null) {
            synchronized (CustomKafkaProducer.class) {
                if (pro == null) {
                    pro = new CustomKafkaProducer();
                }
            }
        }
        return pro;
    }

    public void producerMsg(String key, String value) {
        ProducerRecord<String, String> pr = new ProducerRecord<>(KafkaPreCompute.input_topic, key, value);
        producer.send(pr, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception e) {
                if (metadata == null) {
                    logger.error("[prodcuer send failure] [" + e + "]" + "[" + value + "] ");
                    KafkaConfig.printErrorLog(logger,e);
                }
            }
        });
    }

    public void flush() {
        producer.flush();
    }
}
